package com.org.platform.utils.rabbitmq;

import com.org.platform.utils.CommonUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * @Description
 * @auther dh
 * @date 2022/1/30 21:50
 */
@Slf4j
public class Consumer {
	public static void main(String[] args) throws Exception {
		new InvokeConsumer("Q1").start();
		new InvokeConsumer("Q2").start();
	}

	public static class InvokeConsumer extends Thread {
		private String queueName;

		public InvokeConsumer(String queueName) {
			this.queueName = queueName;
		}

		@Override
		public void run() {
			initConsumer(queueName);
		}
	}

	public static void initConsumer(String queueName) {
		try {
			Channel channel = Producer.getChannel();
			channel.queueDeclare(queueName, true, false, false, null);
			//DefaultConsumer类实现了Consumer接口，通过传入一个频道，
			// 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery
			com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
				public void handleDelivery(String consumerTag, Envelope envelope,
				                           AMQP.BasicProperties properties, byte[] body)
						throws IOException {
					String message = new String(body, "utf-8");
//                Integer expiration = 5000;
//                AMQP.BasicProperties reProperties = new AMQP.BasicProperties.Builder().expiration(expiration.toString()).build();
//                channel.basicPublish("datax-node-jdbc-error-exchange", "re_wait_"+QUEUE_NAME, reProperties, message.getBytes("utf-8"));
					String routingKey = envelope.getRoutingKey();
					String contentType = properties.getContentType();
					log.info("Customer {}, contentType:{},Received:{}", queueName, contentType, message);
					long deliveryTag = envelope.getDeliveryTag();
					channel.basicAck(deliveryTag, false);
				}
			};
			//自动回复队列应答 -- RabbitMQ中的消息确认机制
//  channel.basicConsume(QUEUE_NAME, true, consumer);
			channel.basicQos(5);
			channel.basicConsume(queueName, false, consumer);//这里改为手动确认
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}
}
